package cool.taomu.toolkit.service.centre.kademlia.utils

import com.google.common.base.Optional
import com.google.inject.Inject
import com.google.inject.name.Named
import cool.taomu.toolkit.retry.Retry
import cool.taomu.toolkit.rpc.thrift.service.ThriftConnectionFactory
import cool.taomu.toolkit.service.kademlia.entity.KPeerNode
import cool.taomu.toolkit.service.kademlia.service.PeerRateLimiter
import cool.taomu.toolkit.service.kademlia.thrift.KademliaService
import cool.taomu.toolkit.service.kademlia.thrift.Node
import cool.taomu.toolkit.service.kademlia.utils.NetwrokUtils
import cool.taomu.toolkit.utils.ByteBufferUtils
import java.net.SocketTimeoutException
import java.nio.charset.StandardCharsets
import java.util.concurrent.CompletableFuture
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import java.util.function.Function
import org.apache.commons.pool2.impl.GenericObjectPool
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.thrift.protocol.TCompactProtocol
import org.apache.thrift.transport.TTransport
import org.apache.thrift.transport.layered.TFramedTransport
import org.eclipse.xtend.lib.annotations.Accessors
import org.slf4j.LoggerFactory

@Accessors
class ServiceCentreClientPool {

    val static LOG = LoggerFactory.getLogger(ServiceCentreClientPool);

    @Inject
    @Named("taomu.kademlia.pool.config")
    GenericObjectPoolConfig<TTransport> config;

    @Inject
    ClientExec exec;

    val pool = new ConcurrentHashMap<String, GenericObjectPool<TTransport>>();

    def get(KPeerNode node, KPeerNode localNode) {
        LOG.info("获取一个Thrift服务的连接池对象")
        val pkey = #[node.host, node.port].join("#")
        var result = this.pool.computeIfAbsent(pkey, [
            LOG.info("无 {} 连接池,开始创建", pkey)
            if (!NetwrokUtils.isAvailable(node.host, node.port)) {
                LOG.info("尝试链接 {} 失败，尝试通过代理链接", node)
                NetwrokUtils.setProxyPeer(node)
            }
            if (NetwrokUtils.isAvailable(node.host, node.port)) {
                var factory = new ThriftConnectionFactory(node.host, node.port, 5000, [ socket |
                    LOG.info("通过ping判断连接是否可用")
                    try (var tTransport = new TFramedTransport(socket, Integer.MAX_VALUE)) {
                        var protocol = new TCompactProtocol(tTransport);
                        var client = new KademliaService.Client(protocol);
                        var loaclNode = new Node(localNode.host, localNode.port)
                        var resp = client.ping(loaclNode)
                        var data = ByteBufferUtils.toString(resp.data, StandardCharsets.UTF_8)
                        val pong = data.equals("PONG");
                        if (pong) {
                            LOG.info("ping {} 成功", pkey)
                        }
                        return pong
                    } catch (Exception ex) {
                        LOG.info("连接 {} 异常 {} \n判断是否在池中{}", pkey, ex.message, this.pool.containsKey(pkey));
                        if (this.pool.containsKey(pkey)) {
                            LOG.info("{} 连接出现异常,移除连接", pkey)
                            this.pool.get(pkey).close
                            this.pool.remove(pkey)
                        } else {
                            LOG.info("连接 {} 不在连接池中", pkey)
                        }
                        return false;
                    }
                ]);
                return factory.getObjectPool(config);
            }
        ])
        return Optional.fromNullable(result)
    }

    val rateLimiter = new PeerRateLimiter(10);

    def synchronized <T> Optional<T> peerClient(KPeerNode node, KPeerNode localNode,
        Function<KademliaService.Client, T> func) {
        // 在这里加 Guava 限流
        if (!rateLimiter.getLimiter(node).tryAcquire(1, TimeUnit.SECONDS)) {
            LOG.warn("请求被限流，节点: {}", node);
            return Optional.absent();
        }
        var optional = this.get(node, localNode);
        if (optional.present) {
            var cpool = optional.get()
            var exec = exec.clientExec(cpool, func);
            val result = newArrayList();
           /*  exec.whenComplete([ res, ex |
                if (ex !== null) {
                    LOG.error(ex.message)
                } else {
                    result.add(res)
                }
            ])*/
            result.add(exec.get(1,TimeUnit.MINUTES))
            LOG.info("result.size:{}", result.size)
            if (result.size > 0) {
                return result.get(0)
            }
        }
        return Optional.absent;
    }
}

class ClientExec {
    val static LOG = LoggerFactory.getLogger(ClientExec);

    @Retry(maxAttempts=3, delay=500, exponentialBackoff=true, rateLimit=2.0)
    def <T> CompletableFuture<Optional<T>> clientExec(GenericObjectPool<TTransport> cpool,
        Function<KademliaService.Client, T> func) {
        var socket = cpool.borrowObject
        try(var tTransport = new TFramedTransport(socket, Integer.MAX_VALUE)) {
            var protocol = new TCompactProtocol(tTransport);
            var client = new KademliaService.Client(protocol);
            try {
                var r = CompletableFuture.completedFuture(Optional.fromNullable(func.apply(client)));
                return r
            } catch (SocketTimeoutException stex) {
                throw stex
            }
        } catch (Exception ex) {
            if (socket !== null) {
                LOG.debug("socket标记为无效")
                cpool.invalidateObject(socket) // 标记为无效
            }
            throw ex
        } finally {
            if (socket !== null) {
                cpool.returnObject(socket)
            }
        }
    // return CompletableFuture.completedFuture(Optional.absent);
    }
}
